package com.ccc.demo2;

import org.apache.spark.graphx.EdgeTriplet;
import scala.Tuple2;
import scala.collection.Iterator;
import scala.collection.JavaConversions;
import scala.runtime.AbstractFunction1;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;

//发消息方法
public class MySendMsg extends AbstractFunction1<EdgeTriplet<String, String>, Iterator<Tuple2<Object, MyMessage>>> implements
        Serializable {
    @Override
    public Iterator<Tuple2<Object, MyMessage>> apply(EdgeTriplet<String, String> triplet) {
        List<Tuple2<Object, MyMessage>> list = new ArrayList<>();

        long srcId = triplet.srcId();
        String dstAttr = triplet.dstAttr();
        MyMessage srcMyMessage = null;
        if(dstAttr.equals("black")){
            srcMyMessage = new MyMessage(1, 0);
        }else {
            srcMyMessage = new MyMessage(0, 1);
        }
        list.add(new Tuple2<>(srcId, srcMyMessage));


        long dstId = triplet.dstId();
        String srcAttr = triplet.srcAttr();
        MyMessage dstMyMessage = null;
        if(srcAttr.equals("black")){
            dstMyMessage = new MyMessage(1, 0);
        }else {
            dstMyMessage = new MyMessage(0, 1);
        }
        list.add(new Tuple2<>(dstId, dstMyMessage));
        return JavaConversions.asScalaIterator(list.iterator());
    }
}
